package cn.doitedu.api;


import beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

import java.util.Map;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/22
 * @Desc: 学大数据，上多易教育
 * <p>
 * 从kafka中获取如下用户行为数据:
 * {"uid":1,"timestamp":12309347,"event_id":"A"}
 * {"uid":1,"timestamp":12309347,"event_id":"B"}
 * {"uid":3,"timestamp":12309347,"event_id":"A"}
 * <p>
 * 计算如下需求：
 * 用户，在最近的 5分钟内，发生行为序列： [A C D]的次数 ,输出： {"uid":1,"result_type":1,"seq_cnt":2}
 * 用户，截止到当前，各类事件分别发生的次数,输出: {"uid":1,"result_type":2,"A":3,"B":4,"C":6}
 * <p>
 * 最终结果，要求写回 kafka
 **/

public class _12_Exercise_MapState_ListState {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        // 加载kafka的数据，得到数据流
        KafkaSource<String> source = KafkaSource
                .<String>builder()
                .setTopics("doit46")
                .setBootstrapServers("doitedu:9092")
                .setGroupId("g2")
                .setClientIdPrefix("doit46-")
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "ss");


        // 将json解析成UserAction对象
        SingleOutputStreamOperator<UserAction> mapped = stream.map(new MapFunction<String, UserAction>() {
            @Override
            public UserAction map(String value) throws Exception {
                return JSON.parseObject(value, UserAction.class);
            }
        });

        // 按用户id分组
        KeyedStream<UserAction, Integer> keyedStream = mapped.keyBy(new KeySelector<UserAction, Integer>() {
            @Override
            public Integer getKey(UserAction userAction) throws Exception {
                return userAction.getUid();
            }
        });

        keyedStream.process(new KeyedProcessFunction<Integer, UserAction, String>() {

            ListState<UserAction> actionListState;
            MapState<String, Integer> mapState;

            @Override
            public void open(Configuration parameters) throws Exception {


                RuntimeContext runtimeContext = getRuntimeContext();

                // 获取一个list状态，用来存储用户的历次行为
                this.actionListState = runtimeContext.getListState(new ListStateDescriptor<UserAction>("actions", UserAction.class));

                // 获取一个map状态，用来存储用户的各类行为的次数
                this.mapState = runtimeContext.getMapState(new MapStateDescriptor<String, Integer>("event_count", String.class, Integer.class));


            }

            @Override
            public void processElement(UserAction userAction, KeyedProcessFunction<Integer, UserAction, String>.Context ctx, Collector<String> out) throws Exception {

                // 把当前接收的用户行为，放入行为记录list状态
                actionListState.add(userAction);

                // 遍历list状态，统计ACD序列的次数
                Iterable<UserAction> userActions = actionListState.get();
                int seqCount = eventSeqFind(userActions);
                // 输出行为序列统计结果
                JSONObject res1 = new JSONObject();
                // {"uid":1,"result_type":1,"seq_cnt":2}
                res1.put("uid",userAction.getUid());
                res1.put("result_type",1);
                res1.put("seq_cnt",seqCount);
                out.collect(res1.toJSONString());



                // 根据当前接收的用户行为，去 事件次数map状态中找到此前的次数去增加
                String eventId = userAction.getEvent_id();
                // 获取此前该行为发生的次数
                Integer eventCount = mapState.get(eventId);
                int newCount = eventCount == null ? 1 : eventCount + 1;
                // 把新的次数，更新到状态中
                mapState.put(eventId,newCount);

                JSONObject res2 = new JSONObject();
                // 遍历mapState，取出事件->次数，放入jsonObject
                Iterable<Map.Entry<String, Integer>> entries = mapState.entries();
                for (Map.Entry<String, Integer> entry : entries) {
                    String eid = entry.getKey();
                    Integer eCount = entry.getValue();
                    res2.put(eid,eCount);
                }

                // 输出结果   {"uid":1,"result_type":2,"A":3,"B":4,"C":6}
                res2.put("uid",userAction.getUid());
                res2.put("result_type",2);
                out.collect(res2.toJSONString());


            }


            @Override
            public void close() throws Exception {
            }


            /**
             * 根据一个事件列表，去统计指定行为序列的发生次数
             * @param userActions
             * @return
             */
            public int eventSeqFind(Iterable<UserAction> userActions) {
                // G A B C A A C D X A B Y U C D T  找 A C D的次数
                String[] targetSeq = {"A","C","D"};
                int i = 0; // 正在找第几个
                int cnt = 0; // 一共找到了几次

                for (UserAction userAction : userActions) {

                    if(userAction.getEvent_id().equals(targetSeq[i])){
                        i++;
                        // 如果i到3了，说明整个序列完整地找到了一次；i要重置回0；总次数+1
                        if(i == 3) {
                            i = 0;
                            cnt++;
                        }
                    }
                }

                return cnt;
            }

        });


    }


}
